package com.taotao.delayjob.core;

import com.taotao.delayjob.core.constant.RetryConstant;
import com.taotao.delayjob.core.model.RetryInst;
import lombok.extern.slf4j.Slf4j;

import javax.sql.DataSource;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * 重试任务调度
 * </p>
 *
 * @author fanbeibei
 * @version 1.0.0
 * @date 2024/7/17 10:33
 */
@Slf4j
public class RetryTaskScheduler {
    private DataSource dataSource;
    private RetryTaskHelper retryTaskHelper;


    public RetryTaskScheduler(DataSource dataSource, RetryTaskHelper retryTaskHelper) {
        this.dataSource = dataSource;
        this.retryTaskHelper = retryTaskHelper;
    }

    private volatile boolean scheduleThreadToStop = false;

    private Thread scheduleThread;

    public void start() {

        scheduleThread = new Thread(new Runnable() {
            @Override
            public void run() {
                start0();
            }
        });

        scheduleThread.setDaemon(true);
        scheduleThread.setName("retryScheduler#scheduleThread");
        scheduleThread.start();

    }

    protected void start0() {
        try {
            //  每隔1分钟扫一次
            TimeUnit.MINUTES.sleep(1);
        } catch (InterruptedException e) {
            if (!scheduleThreadToStop) {
                log.error(e.getMessage(), e);
            }
        }
        log.info(">>>>>>>>> init retry scheduler success.");


        while (!scheduleThreadToStop) {

            try {
                //  每隔1分钟扫一次
                TimeUnit.MINUTES.sleep(1);
            } catch (InterruptedException e) {
                if (!scheduleThreadToStop) {
                    log.error(e.getMessage(), e);
                }
            }

            Connection conn = null;
            Boolean connAutoCommit = null;
            PreparedStatement preparedStatement = null;

            boolean preReadSuc = true;
            try {

                // 获取锁
                conn = dataSource.getConnection();
                connAutoCommit = conn.getAutoCommit();
                conn.setAutoCommit(false);
                preparedStatement = conn.prepareStatement("select lock_id from retry_lock where lock_id = " + RetryConstant.RETRY_LOCK_ID + " for update");
                preparedStatement.execute();


                // 扫描执行重试逻辑
                List<RetryInst> retryList = retryTaskHelper.selectForRetryInst(System.currentTimeMillis());
                log.info("retryList={}", retryList);
                if (null == retryList || retryList.isEmpty()) {
                    continue;
                }


                for (RetryInst r : retryList) {
                    retryTaskHelper.triggerRetry(r);
                }


            } catch (Exception e) {
                if (!scheduleThreadToStop) {
                    log.error(">>>>>>>>>>> retryScheduleHelper#scheduleThread error:{}", e);
                }
            } finally {
                // commit
                if (conn != null) {
                    try {
                        conn.commit();
                    } catch (SQLException e) {
                        if (!scheduleThreadToStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        conn.setAutoCommit(connAutoCommit);
                    } catch (SQLException e) {
                        if (!scheduleThreadToStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                    try {
                        conn.close();
                    } catch (SQLException e) {
                        if (!scheduleThreadToStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }

                // close PreparedStatement
                if (null != preparedStatement) {
                    try {
                        preparedStatement.close();
                    } catch (SQLException e) {
                        if (!scheduleThreadToStop) {
                            log.error(e.getMessage(), e);
                        }
                    }
                }
            }


        }

        log.info(">>>>>>>>>>> retryScheduleHelper#scheduleThread stop");
    }


    public void destroy() {
        this.scheduleThreadToStop = true;
        try {
            TimeUnit.SECONDS.sleep(1);  // wait
        } catch (InterruptedException e) {
            log.error(e.getMessage(), e);
        }
        if (scheduleThread.getState() != Thread.State.TERMINATED) {
            // interrupt and wait
            scheduleThread.interrupt();
            try {
                scheduleThread.join();
            } catch (InterruptedException e) {
                log.error(e.getMessage(), e);
            }
        }
    }


}
